package com.jumi.microservice.rocketmq;

import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.bean.ConsumerBean;
import com.aliyun.openservices.ons.api.bean.Subscription;
import com.jumi.microservice.common.exception.GlobalExceptionHandler;
import com.jumi.microservice.config.RocketMqConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

/**
 * @Description 消费者
 * @ClassName RocketMqConsumer
 * @Author Abel Yang
 * @Date 2020/10/24/024 10:22
 **/
@Configuration
public class MessageMqConsumer {

    private static final Logger log = LoggerFactory.getLogger(GlobalExceptionHandler.class);

    @Resource
    private RocketMqConfig rocketMqConfig;

    @Resource
    private MessageMqListener messageListener;

    @Bean(initMethod = "start", destroyMethod = "shutdown")
    public ConsumerBean buildConsumer() {
        log.info("-----------Apache RocketMQ Consumer creating ----------------------");
        ConsumerBean consumerBean = new ConsumerBean();
        //配置文件
        Properties properties = rocketMqConfig.getMqProperties();
        properties.setProperty(PropertyKeyConst.GROUP_ID, rocketMqConfig.getGroup());
        //将消费者线程数固定为20个 20为默认值
        properties.setProperty(PropertyKeyConst.ConsumeThreadNums, "20");
        consumerBean.setProperties(properties);
        //订阅关系
        Map<Subscription, MessageListener> subscriptionTable = new HashMap<Subscription, MessageListener>();
        Subscription subscription = new Subscription();
        subscription.setTopic(rocketMqConfig.getTopic());
//        subscription.setExpression(rocketMqConfig.getTag());
        subscriptionTable.put(subscription, messageListener);
        //订阅多个topic如上面设置

        consumerBean.setSubscriptionTable(subscriptionTable);
        log.info("-----------Apache RocketMQ Consumer created ----------------------");
        return consumerBean;
    }
}
